package com.mcy.springbatch.reader;

import com.mcy.springbatch.entity.SysUser;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.item.*;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.core.io.FileSystemResource;
import org.springframework.stereotype.Component;

/**
 * 利用ItemStreamReader从文件读取数据，处理异常及重启读取，续跑任务
 * @author : Create By mcy
 * @date : 2019/8/25 14:43:53
 */
@Slf4j
@Component
public class RestartItemStreamReader implements ItemStreamReader<SysUser> {

    private FlatFileItemReader<SysUser> reader = new FlatFileItemReader<>();
    private Long currentLine = 0L;
    private boolean restart = false;
    private ExecutionContext context;

    public RestartItemStreamReader() {
        reader.setResource(new FileSystemResource("/Users/calon/Desktop/userdata.txt"));
        reader.setLineMapper(new DefaultLineMapper<SysUser>() {{
            setLineTokenizer(new DelimitedLineTokenizer("|") {{
                setNames(new String[]{"id", "loginName", "userName", "idCard", "phone", "sex", "address", "email", "status"});
            }});
            setFieldSetMapper(new BeanWrapperFieldSetMapper<SysUser>() {{
                setTargetType(SysUser.class);
            }});
        }});
    }


    @Override
    public void open(ExecutionContext executionContext) throws ItemStreamException {
        this.context = executionContext;
        if (executionContext.containsKey("currentLine")) {
            this.currentLine = executionContext.getLong("currentLine");
            this.restart = true;
        } else {
            this.currentLine = 0L;
            executionContext.put("currentLine", this.currentLine);
            log.info("开始从第几行读数据：{}", this.currentLine + 1);
        }
    }

    @Override
    public SysUser read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
        this.currentLine++;
        if (restart) {
            reader.setLinesToSkip(this.currentLine.intValue() - 1);
            restart = false;
            log.info("从哪行开始读：{}", this.currentLine);
        }
        reader.open(this.context);
        SysUser user = reader.read();
        return user;
    }

    @Override
    public void update(ExecutionContext executionContext) throws ItemStreamException {
        executionContext.put("currentLine", this.currentLine);
        log.info("保存进度：{}", this.currentLine);
    }

    @Override
    public void close() throws ItemStreamException {

    }
}
